package org.example.batch.task;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.Setter;
import org.example.batch.dao.BatchSubTaskDao;
import org.example.batch.dao.BatchTerResultDao;
import org.example.batch.dao.InputTableInfoDao;
import org.example.batch.pojo.*;
import org.example.batch.util.constant.BatchTerSubStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

import static org.example.batch.util.tools.ToolsUtil.postUrl;

/**
 * 子任务线程核心类
 *
 * @author hero生仔
 * @date 2025/7/26
 */
@Component
@Scope(scopeName = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class BatchRunRecordTask implements Runnable {

    private final static Logger LOGGER = LoggerFactory.getLogger(BatchRunRecordTask.class);

    @Autowired
    private BatchSubTaskDao batchSubTaskDao;

    @Autowired
    private InputTableInfoDao inputTableInfoDao;

    @Autowired
    private BatchTerResultDao batchTerResultDao;

    /**
     * 事务模板
     */
    @Autowired
    private TransactionTemplate transactionTemplate;

    /**
     * 每次处理的条数
     */
    @Value("${ter.offset}")
    private int offset;

    /**
     * 虚拟的url；仅用做测试
     */
    @Value("${ter.url}")
    private String url;

    @Setter
    private BatchSubTable batchSubTableThread;

    @Setter
    private CountDownLatch countDownLatch;

    @Setter
    private BatchMainTable batchMainTable;

    @Override
    public void run() {
        // 只有一个线程才能进入处理数据的循环
        boolean flag = batchSubTaskDao.updateSubTableStatus(batchSubTableThread) > 0;
        while (flag) {
            // 默认终止循环
            flag = false;
            // 1. 读取输入表信息
            List<InputTableInfo> inputTableInfo = inputTableInfoDao.queryInputInfo(appendQueryInfo(batchSubTableThread, offset));
            int size = inputTableInfo.size();

            JSONArray jsonArray;
            BatchSubTablePojo batchSubTablePojo = new BatchSubTablePojo();
            if (size > 0) {
                int step = size > offset ? size - 1 : size;
                // 2. 调取外部系统
                jsonArray = callSystem(inputTableInfo.subList(0, step));
                // 3. 处理响应报文
                batchSubTablePojo = handleResponseJsonArray(jsonArray, inputTableInfo.subList(0, step));
                if (BatchTerSubStatus.CALL_FAIL.name().equalsIgnoreCase(batchSubTablePojo.getStatus())) {
                    batchSubTableThread.setStatus(BatchTerSubStatus.CALL_FAIL.name());
                } else if (BatchTerSubStatus.RCD_FAIL.name().equalsIgnoreCase(batchSubTablePojo.getStatus())) {
                    batchSubTableThread.setStatus(BatchTerSubStatus.RCD_FAIL.name());
                } else {
                    BatchTerSubStatus status = size > offset ? BatchTerSubStatus.RUNNING : BatchTerSubStatus.SUCCESS;
                    batchSubTableThread.setStatus(status.name());
                    batchSubTableThread.setAlreadyRunningCount(batchSubTableThread.getAlreadyRunningCount() + step);
                    // 记录lastQueryKey
                    batchSubTableThread.setLastQueryKey(Optional.ofNullable(inputTableInfo.get(size - 1)).orElse(new InputTableInfo()).getId());
                }
            }

            // 4. 持久化处理：再次查询子任务表，状态为RUNNING
            String status = batchSubTaskDao.queryBatchSubTaskStatus(batchSubTableThread.getSubTaskId());
            if (BatchTerSubStatus.RUNNING.name().equalsIgnoreCase(status)) {
                // 更新子任务状态 和 持久化跑批调额数据 在一个事务中
                saveSubTableAndTerResultData(batchSubTablePojo);
                // 如果当前的子任务状态仍然是RUNNING，则继续循环
                if (BatchTerSubStatus.RUNNING.name().equalsIgnoreCase(batchSubTableThread.getStatus())) {
                    flag = true;
                }
            }
        }

        countDownLatch.countDown();
    }

    /**
     * 更新子任务状态 和 持久化跑批调额数据 在一个事务中
     *
     * @param finalBatchSubTablePojo 中间变量
     */
    private void saveSubTableAndTerResultData(BatchSubTablePojo finalBatchSubTablePojo) {
        transactionTemplate.execute(transactionStatus -> {
            batchSubTableThread.setErrorMsg(finalBatchSubTablePojo.getErrorMsg());
            batchSubTaskDao.updateSubTableInfo(batchSubTableThread);
            boolean flag = BatchTerSubStatus.RUNNING.name().equalsIgnoreCase(batchSubTableThread.getStatus()) ||
                    BatchTerSubStatus.SUCCESS.name().equalsIgnoreCase(batchSubTableThread.getStatus());
            if (flag) {
                try {
                    // 异常，回滚事务
//                    int i = 1 / 0;
                    batchTerResultDao.batchInsertTerResult(finalBatchSubTablePojo.getBatchTerResultList());
                } catch (Exception e) {
                    LOGGER.error("插入记录信息为：{}，插入异常信息：{}", JSONObject.toJSONString(finalBatchSubTablePojo.getBatchTerResultList()), e.getMessage());
                    // 回滚事务
                    transactionStatus.setRollbackOnly();
                    // 释放锁，防止死锁
                    countDownLatch.countDown();
                    throw e;
                }
            }
            // 提交事务
            return true;
        });
    }

    /**
     * 处理响应报文
     *
     * @param jsonArray       请求结果
     * @param inputTableInfos 输入表
     * @return 中间变量
     */
    private BatchSubTablePojo handleResponseJsonArray(JSONArray jsonArray, List<InputTableInfo> inputTableInfos) {
        BatchSubTablePojo batchSubTablePojo = new BatchSubTablePojo(new ArrayList<>(), BatchTerSubStatus.SUCCESS.name(), "");
        if (jsonArray == null || jsonArray.isEmpty() || jsonArray.size() != inputTableInfos.size()) {
            batchSubTablePojo.setErrorMsg("调用外部系统失败");
            batchSubTablePojo.setStatus(BatchTerSubStatus.CALL_FAIL.name());
            return batchSubTablePojo;
        }

        List<BatchTerResult> batchTerResultList;
        batchTerResultList = jsonArray.stream().map(obj -> (JSONObject) obj)
                .filter(jsonResult -> {
                    JSONObject dcsResult = Optional.ofNullable(jsonResult.getJSONObject("dcsResult")).orElse(new JSONObject());
                    String lmtChkRst = dcsResult.getString("lmt_chk_rst");
                    BigDecimal newLmt = dcsResult.getBigDecimal("new_lmt");
                    BigDecimal oldLmt = dcsResult.getBigDecimal("old_lmt");
                    return ("INC".equalsIgnoreCase(lmtChkRst) && newLmt.compareTo(oldLmt) > 0)
                            || ("DEC".equalsIgnoreCase(lmtChkRst) && newLmt.compareTo(oldLmt) < 0)
                            || ("NOCG".equalsIgnoreCase(lmtChkRst) && newLmt.compareTo(oldLmt) == 0);
                }).map(jsonResult -> {
                    BatchTerResult batchTerResult = new BatchTerResult();
                    JSONObject dcsResult = Optional.ofNullable(jsonResult.getJSONObject("dcsResult")).orElse(new JSONObject());

                    batchTerResult.setSubTaskId(batchSubTableThread.getSubTaskId());
                    batchTerResult.setMainTaskId(batchSubTableThread.getMainTaskId());
                    batchTerResult.setDwSnshDt(batchMainTable.getDwSnshDt());
                    batchTerResult.setCltNum(dcsResult.getString("clt_num"));
                    batchTerResult.setRootLmtId(dcsResult.getString("root_lmt_id"));
                    batchTerResult.setLmtId(dcsResult.getString("lmt_id"));
                    batchTerResult.setLmtChkRst(dcsResult.getString("lmt_chk_rst"));
                    batchTerResult.setOldLmt(dcsResult.getBigDecimal("old_lmt"));
                    batchTerResult.setNewLmt(dcsResult.getBigDecimal("new_lmt"));
                    batchTerResult.setRateChkRst(dcsResult.getString("rate_chk_rst"));
                    batchTerResult.setOldRate(dcsResult.getBigDecimal("old_rate"));
                    batchTerResult.setNewRate(dcsResult.getBigDecimal("new_rate"));
                    return batchTerResult;
                }).collect(Collectors.toList());

        // 如果跑批调额结果表数据和输入表数据不一致，则单笔校验失败
        if (batchTerResultList.size() != inputTableInfos.size()) {
            batchSubTablePojo.setErrorMsg("单笔记录校验失败");
            batchSubTablePojo.setStatus(BatchTerSubStatus.RCD_FAIL.name());
        } else {
            batchSubTablePojo.setBatchTerResultList(batchTerResultList);
        }
        return batchSubTablePojo;
    }

    /**
     * 调外部系统
     *
     * @param inputTableInfos 输入表
     * @return 响应报文
     */
    private JSONArray callSystem(List<InputTableInfo> inputTableInfos) {
        JSONArray jsonArray = new JSONArray();
        List<JSONObject> inputObjectList = convertToJsonObj(inputTableInfos);
        try {
            jsonArray = postUrl(url, inputObjectList);
        } catch (InterruptedException e) {
            LOGGER.error("重试机制中断异常：{}", e.getMessage());
        }
        return jsonArray;
    }

    /**
     * 转换对象（仅模拟作用，这些字段又不是真正作为请求外部系统的报文）
     *
     * @param inputTableInfos 输入表信息
     * @return 输入表的JSONObject集合
     */
    private List<JSONObject> convertToJsonObj(List<InputTableInfo> inputTableInfos) {
        JSONObject jsonObject = new JSONObject();
        return inputTableInfos.stream().map(inputTableInfo -> {
            jsonObject.put("LMT_ID", inputTableInfo.getLmtId());
            jsonObject.put("COMMENT", inputTableInfo.getComment());
            return jsonObject;
        }).collect(Collectors.toList());
    }

    /**
     * 查询输入表添加基本信息
     *
     * @param batchSubTableThread 子线程
     * @param offsets             偏移量
     * @return 输入表信息
     */
    private Map<String, Object> appendQueryInfo(BatchSubTable batchSubTableThread, int offsets) {
        Map<String, Object> params = new HashMap<>();
        params.put("tableName", batchSubTableThread.getInputTableName());
        params.put("lastQueryKey", batchSubTableThread.getLastQueryKey());
        params.put("end", batchSubTableThread.getInputTableEnd());
        params.put("offset", offsets + 1);
        return params;
    }
}
